/*
 * 文件名：PollableSourceRunner.java
 * 版权：亚信联创版权所有
 * 描述：TODO
 * 修改人：napo
 * 修改时间：2014-12-16
 * 修改内容：TODO
 */
package com.ailk.bigdata.etl.realstream.server.protocal.realmq;

import java.util.concurrent.atomic.AtomicBoolean;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ailk.bigdata.etl.realstream.server.protocal.realmq.lifrcycle.LifecycleAware;
import com.ailk.bigdata.etl.realstream.server.protocal.realmq.lifrcycle.LifecycleState;

/**
 * @description TODO
 * @author [xuwei3]
 * @version [版本号,2014-12-16]
 * @see [相关类/方法]
 * @since [产品/模块版本]
 */
public class PollableSourceRunner implements LifecycleAware
{
	private static final Logger logger = LoggerFactory.getLogger(PollableSourceRunner.class);
	private static final long maxBackoffSleep = 5000;
	private PollableSource source;
	private AtomicBoolean shouldStop;
	private PollingRunner runner;
	private Thread runnerThread;
	private LifecycleState lifecycleState;

	public PollableSourceRunner()
	{
		shouldStop = new AtomicBoolean();
		lifecycleState = LifecycleState.IDLE;
	}

	@Override
	public void start()
	{
		PollableSource source = (PollableSource) getSource();
		source.start();
		runner = new PollingRunner();
		runner.source = source;
		runner.shouldStop = shouldStop;
		runnerThread = new Thread(runner);
		runnerThread.setName(getClass().getSimpleName() + "-" + source.getClass().getSimpleName());
		runnerThread.start();
		lifecycleState = LifecycleState.START;
	}

	@Override
	public void stop()
	{
		runner.shouldStop.set(true);
		try
		{
			runnerThread.interrupt();
			runnerThread.join();
		}
		catch (InterruptedException e)
		{
			logger.warn("Interrupted while waiting for polling runner to stop. Please report this.", e);
			Thread.currentThread().interrupt();
		}
		PollableSource source = getSource();
		source.stop();
		lifecycleState = LifecycleState.STOP;
	}

	@Override
	public LifecycleState getLifecycleState()
	{
		return lifecycleState;
	}

	public static class PollingRunner implements Runnable
	{
		private PollableSource source;
		private AtomicBoolean shouldStop;

		@Override
		public void run()
		{
			logger.debug("Polling runner starting. Source:{}", source);
			while (!shouldStop.get())
			{
				try
				{
					if (source.process().equals(PollableSource.Status.BACKOFF))
					{
						Thread.sleep(maxBackoffSleep);
					}
				}
				catch (InterruptedException e)
				{
					logger.info("Source runner interrupted. Exiting");
				}
				catch (Exception e)
				{
					logger.error("Unhandled exception, logging and sleeping for " + maxBackoffSleep + "ms", e);
					try
					{
						Thread.sleep(maxBackoffSleep);
					}
					catch (InterruptedException ex)
					{
						Thread.currentThread().interrupt();
					}
				}
			}
		}
	}

	public PollableSource getSource()
	{
		return this.source;
	}

	public void setSource(PollableSource source)
	{
		this.source = source;
	}
}
